
use std::{io::Write, thread::spawn, time::Duration};

use clap::Parser;
use serde::{Deserialize, Serialize};

use wtask_base::{
    error::{WError, WResult},
    export::{KillReceiver, KillSender, broadcast_channel, kill_channel, log_debug, log_error, log_info, select, tokio_sleep, yield_now},
    machine::{WSpawn, get_machine_name, get_machine_token, runtime_init}, tcp::{WTcpStream, stream_broadchannel_exchange_skey, stream_exchange_dkey, stream_read_raw, tcp_client_session_start, tcp_listen, tcp_server_session_start}
};

use crate::{basic::Empty, func::{stdin_read_line, write_log}, utils::{AppTrait, StreamStatus}};



#[derive(Debug, Clone, Serialize, Deserialize, Parser)]
pub struct AppRoom {
    pub(super) port: u32,

    #[arg(long, default_value = "1024")]
    pub(super) capacity: usize,
}


impl AppRoom {
    pub(super) async fn master(&self, server_token: &str, mut kill_rx: KillReceiver<bool>) -> WResult<KillSender<bool>> {
        // listen
        let port = self.port;
        let capacity = self.capacity;
        let server_token_clone = server_token.to_string();
        let (kill_tx_member, mut kill_rx_member) = kill_channel(false);
        let _t: std::thread::JoinHandle<Result<(), WError>> = spawn(move || {
            let runtime  = runtime_init(8)?;
            let _t: Result<(), WError> = runtime.block_on(async move {
                // log_debug!("Start room ...");
                let listener = tcp_listen(port, "Room")?;
                // log_debug!("Room started");
                let mut wspawn = WSpawn::default();
                let (kill_tx_pause, mut kill_rx_pause) = wspawn.catch_pause_signal(async move {});
                let (sender, _) = broadcast_channel::<Vec<u8>>(capacity);
                // log_debug!("Room set channel");
                loop {
                    select! {
                        Ok((stream, _)) = listener.accept() => {
                            // log_debug!("room get a connect ...");
                            let server_token = server_token_clone.clone();
                            let sender_clone = sender.clone();
                            let receiver_clone = sender.subscribe();
                            wspawn.spawn(async move {
                                let stream = stream.into_split();
                                let session_token = tcp_server_session_start(&stream, &server_token).await;
                                if let Ok(session_token) = session_token {
                                    log_debug!("room connected session: {session_token}");
                                    let _ = stream_broadchannel_exchange_skey(&stream, sender_clone, receiver_clone, &session_token).await;
                                } else {
                                    log_error!("Master session connect error");
                                }
                            });
                        }
                        _ = kill_rx_pause.changed() => {
                            log_info!("Shutting room {} ...", port);
                            break;
                        }
                        _ = kill_rx_member.changed() => {
                            let _ = kill_tx_pause.send(true);
                            log_info!("Shutting room {}, kill by member ...", port);
                            break;
                        }
                        _ = kill_rx.changed() => {
                            let _ = kill_tx_pause.send(true);
                            log_info!("Shutting room {}, kill by parent ...", port);
                            break;
                        }
                    }
                    yield_now().await;
                }
                Ok(())
            });
            runtime.shutdown_background();
            Ok(())
        });
        tokio_sleep(Duration::from_secs(1)).await;
        Ok(kill_tx_member)
    }

    async fn conn_master(self, server_token: &str, session_token: &str, stream_session: &WTcpStream) -> WResult<()> {
        // exchange
        log_debug!("room get a session: {session_token}");
        let stream_master = tcp_client_session_start(
            "127.0.0.1",
            self.port,
            server_token,
            session_token,
        ).await?;
        log_debug!("room connect server: {session_token}");
        let _ = stream_exchange_dkey(&stream_master, stream_session, session_token).await;
        Ok(())
    }
}




impl AppTrait for AppRoom {
    type Status = ();
    type Res = Empty;

    async fn client_handle<M: FnMut(bool, String) + Send + Sync>(self, stream: &WTcpStream, key: &str, _msg_func: M, _status: Option<Self::Status>) -> WResult<Empty> {
        let (stream_read, stream_write) = stream;
        let id = {
            let mut buf = String::new();
            print!("Please input user name: ");
            std::io::stdout().flush().unwrap();
            let _ = std::io::stdin().read_line(&mut buf);
            let id = get_machine_token()?;
            format!("{}@{}: ", buf.trim(), get_machine_name(&id))
        };
        log_info!("UserID: {}", id);
        let a = stdin_read_line(stream_write.as_ref(), key, &id);
        let key_clone = key.to_string();
        let b = async move {
            loop {
                log_debug!("cmd client wait remote ...");
                let v = stream_read_raw(stream_read.as_ref(), &key_clone).await;
                if let Ok(data) = v {
                    write_log(data);
                } else {
                    log_debug!("cmd client read error: {:?}", v.err().unwrap());
                    break;
                }
            }
        };
        select! {
            _ = a => {},
            _ = b => {},
        }
        Ok(().into())
    }
    
    async fn server_handle(self, server_token: &str, session_token: &str, stream_session: &WTcpStream, stream_status: StreamStatus) -> WResult<()> {
        let port = self.port;
        let n = stream_status.read().await.2.get(&port).cloned();
        if let Some(n) = n {
            // 服务已经启动，增加次数
            stream_status.write().await.2.insert(port, n+1);
        } else {
            // 第一次需要启动
            let kill_rx = stream_status.read().await.0.clone();
            stream_status.write().await.2.insert(port, 1);
            let kill_tx = self.master(server_token, kill_rx).await?;
            stream_status.write().await.1.insert(port, kill_tx);
            // log_debug!("room {} give a killer", port);
        }
        let c = self.conn_master(server_token, session_token, stream_session).await;
        if let Err(e) = c {
            log_error!("Room error: {}", e);
        }
        let mut s = stream_status.write().await;
        let n = s.2.get_mut(&port);
        match n {
            Some(1) | None | Some(0) => {
                log_info!("The last person close room {}", port);
                let _ = s.2.remove(&port);
                let kill_tx = s.1.remove(&port).unwrap();
                let _ = kill_tx.send(true);
            },
            Some(n) => {
                *n -= 1;
                log_info!("Someone left room {}, left {}", port, *n);
            }
        }
        Ok(())
    }
}
